package com.atguigu.gmall.realtime.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.util.List;

/**
 * @author Felix
 * @date 2023/1/8
 * 查询维度的工具类
 */
public class DimUtil {

    public static JSONObject getDimInfo(String tableName, String id) {
        return getDimInfo(tableName,Tuple2.of("id",id));
    }
    /**
     * @param tableName
     * @param columnNameAndValues
     * @return
     * 优化：旁路缓存
     * 思路
     *      先从缓存中获取维度数据，如果缓存中能够找到要关联的维度，那么直接将缓存中的维度作为返回值返回(缓存命中)；
     *      如果在缓存中没有找到要关联的维度，那么发送请求，到Phoenix表中去查询维度数据，并且将查询出来的维度数据
     *      放到缓存中缓存起来，方便下次查询使用
     * 选型
     *      redis       性能好 维护性好    √
     *      状态         性能好 维护性差
     * 关于Redis的设置
     *      类型：     String
     *      Key:      dim:维度表名:主键1_主键2
     *      expire：  1day
     * 注意
     *      一旦Phoenix维度表中的数据 发生了变化，将缓存的数据清除
     */
    public static JSONObject getDimInfo(String tableName, Tuple2<String, String>... columnNameAndValues) {
        //拼接从Redis中查询的数据key
        StringBuilder redisKey = new StringBuilder("dim:"+tableName.toLowerCase()+":");
        //拼接查询语句
        StringBuilder selectSql = new StringBuilder("select * from " + GmallConfig.DIM_SCHEMA + "." + tableName + " where ");
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + " = '" + columnValue + "'");
            redisKey.append(columnValue);
            if(i < columnNameAndValues.length - 1){
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }

        Jedis jedis = null;
        String dimStr = null;
        JSONObject dimJsonObj = null;

        try {
            jedis = RedisUtil.getJedis();
            dimStr = jedis.get(redisKey.toString());
        } catch (Exception e) {
            e.printStackTrace();
        }

        //判断是否能够从Redis缓存中获取到维度数据
        if(StringUtils.isNotEmpty(dimStr)){
            //如果缓存中能够找到要关联的维度，那么直接将缓存中的维度作为返回值返回(缓存命中)
            dimJsonObj = JSON.parseObject(dimStr);
        }else{
            //如果在缓存中没有找到要关联的维度，那么发送请求，到Phoenix表中去查询维度数据
            System.out.println("从Phoenix表中查询维度的SQL:" + selectSql);
            //底层还是调用的PhoenixUtil.queryList
            List<JSONObject> resList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
            if (resList != null && resList.size() > 0) {
                //因为我们是按照维度的主键进行查询，所以如果查到维度数据，那么维度数据之后有一条
                dimJsonObj = resList.get(0);
                //将查询的结果放到缓存中缓存起来，方便下次查询使用
                if(jedis != null){
                    jedis.setex(redisKey.toString(),3600*24,dimJsonObj.toJSONString());
                }
            } else {
                System.out.println("从Phoenix维度表中没有查到对应的维度数据");
            }
        }

        //释放资源
        if(jedis != null){
            System.out.println("~~关闭Jedis客户端~~");
            jedis.close();
        }

        return dimJsonObj;
    }

    //Phoenix维度表数据发生变化之后，将缓存的数据清除掉
    public static void deleteCached(String tableName,String id){
        String redisKey = "dim:"+tableName.toLowerCase()+":" + id;
        Jedis jedis = null;
        try {
            jedis = RedisUtil.getJedis();
            jedis.del(redisKey);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(jedis != null){
                System.out.println("~~删除维度之后，关闭Jedis客户端~~");
                jedis.close();
            }
        }
    }

    //从Phoenix表中查询维度数据
    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String, String>... columnNameAndValues) {
        //拼接查询语句
        StringBuilder selectSql = new StringBuilder("select * from " + GmallConfig.DIM_SCHEMA + "." + tableName + " where ");
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + " = '" + columnValue + "'");
            if(i < columnNameAndValues.length - 1){
                selectSql.append(" and ");
            }
        }
        System.out.println("从Phoenix表中查询维度的SQL:" + selectSql);
        //底层还是调用的PhoenixUtil.queryList
        List<JSONObject> resList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
        JSONObject dimJsonObj = null;
        if (resList != null && resList.size() > 0) {
            //因为我们是按照维度的主键进行查询，所以如果查到维度数据，那么维度数据之后有一条
            dimJsonObj = resList.get(0);
        } else {
            System.out.println("从Phoenix维度表中没有查到对应的维度数据");
        }
        return dimJsonObj;
    }

    public static void main(String[] args) {

        // JSONObject dimInfoJsonObj = getDimInfoNoCache("DIM_BASE_trademark", Tuple2.of("id", "1"));
        JSONObject dimInfoJsonObj = getDimInfo("DIM_BASE_trademark", Tuple2.of("id", "1"));
        System.out.println(dimInfoJsonObj);
    }
}
